package pageRank;

import java.io.IOException;
import java.util.Iterator;

import mapred.AbstractCombiner;
import mapred.util.Bytes;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import removeDanglingPage.ReadBytes;

public class PageRank_Combiner extends AbstractCombiner<BytesWritable, BytesWritable> {

	@Override
	protected void combine(BytesWritable key, Iterator<BytesWritable> values,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		double sum = 0;
		while(values.hasNext()) {
			ReadBytes v = new ReadBytes(values.next());
			if(v.size() < 16) {
				sum += v.nextDouble();
			} else {
				output.collect(key, new BytesWritable(v.nextBytes(v.size())));
			}
		}
		output.collect(key, new BytesWritable(Bytes.Double2Bytes(sum)));
	}

//	@Override
//	protected void combine(Text key, Iterator<Text> values,
//			OutputCollector<Text, Text> output, Reporter reporter)
//			throws IOException {
//		double sum = 0;
//		while (values.hasNext()) {
//			Text value = values.next();
//			if (value.toString().startsWith("#")) {
//				sum += Double.parseDouble(value.toString().substring(1).trim());
//			} else {
//				output.collect(key, value);
//			}
//		}
//		output.collect(key, new Text("#" + sum));
//	}

}
